package com.chukun.kafa.action.consumer.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Properties;

public class KafkaNewConsumer {
	
	public static void main(String[] args) {
		
		Properties props = KafkaConsumerPropertiesUtils.props("test");

		// 修改分区分配策略
		ArrayList<String> startegys = new ArrayList<>();
		startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
		props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, startegys);
		// props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
		// 定义consumer 
		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
		
		//消费者订阅的topic, 可同时订阅多个
		consumer.subscribe(Collections.singletonList("kafka-topic-test01"));
		
		while(true) {
			//读取数据，读取超时时间为100ms
			ConsumerRecords<String,String> records = consumer.poll(Duration.ofSeconds(1));
			for(ConsumerRecord<String, String> record:records) {
				System.out.println(record);
			}
		}
	}

}
